package hook;


/**
 * 
 * @author wyl
 * @date 2020/12/09
 */


import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.IOException;
import java.util.Properties;

public class Producer extends Thread {


	private KafkaProducer<String, String> producer = null;
	private ProducerRecord<String, String> record = null;

	public Producer(KafkaProducer<String, String> producer, ProducerRecord<String, String> record) {
		this.producer = producer;
		this.record = record;
	}

	static Properties initConfig() {
		Properties properties = new Properties();
		Properties config = new Properties();

		try {
			config.load(Producer.class.getClassLoader().getResourceAsStream("application.properties"));
		} catch (IOException e) {
			e.printStackTrace();
		}
		String brokerList = config.getProperty("datasource.kafka.url");
		properties.put(org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
		// 设置序列化
		properties.put(org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
				StringSerializer.class.getName());
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		return properties;

	}

	public void run() {
		sendMsgFunc(producer, record);
	}

	static void sendMsgFunc(KafkaProducer<String, String> producer, ProducerRecord<String, String> record) {
		// 异步发送，调用回调函数
		producer.send(record, new Callback() {
			public void onCompletion(RecordMetadata recordMetadata, Exception e) {
				if (null != e) {
					System.out.println("Send message occurs exception.");
				}
				if (null != recordMetadata) {
					System.out.println(String.format("offset:%s,partition:%s", recordMetadata.offset(),
							recordMetadata.partition()));
				}
			}
		});
		try {
			Thread.sleep(20);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
	}

	public static void main(String[] args) {
		ProducerRecord<String, String> record = null;
		Properties configs = Producer.initConfig();

		KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs);
		record = new ProducerRecord<String, String>("test", Integer.toString(1), "poducer's test message");
		Producer producerThread = new Producer(producer, record);
		producerThread.start();

	}

}
